package io.narayana.txdemo.jms;

import javax.jms.*;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.ArrayList;
import java.util.List;

import static io.narayana.txdemo.jms.QueueConst.JMS_QUEUE_JNDI;
import static io.narayana.txdemo.jms.WildFlyJmsQueue.getInitialContext;

/**
 * @author machinly
 * @date 16/4/20
 * @mail machinly@outlook.com
 */
public class WildFlyJmsQueueReceive implements MessageListener, Runnable {
    private static WildFlyJmsQueueReceive wjqr;
    private QueueConnectionFactory qconFactory;
    private InitialContext ic;
    private QueueConnection qcon;
    private QueueSession qsession;
    private QueueReceiver qReceiver;
    private Queue queue;
    private TextMessage msg;
    private boolean quit = false;
    private List<String> msgs = new ArrayList<>();

    public static WildFlyJmsQueueReceive getInstance() {
        if (wjqr == null) {
            wjqr = new WildFlyJmsQueueReceive();
            Thread thread = new Thread(wjqr);
            thread.start();
        }
        return wjqr;
    }


    public static void main(String[] args) throws Exception {
        System.out.println("JMS Ready To Receive Messages (To quit, send a \"quit\" message from QueueSender.class).");
        for (int i = 0; i < 1000; i++) {
            Thread.sleep(1000L);
            List<String> msgs = WildFlyJmsQueueReceive.getInstance().getMsg();
            for (String msg : msgs) {
                System.out.println(msg);
            }
        }
    }

    public List<String> getMsg() {
        List<String> tmpMsgs = null;
        synchronized (msgs) {
            tmpMsgs = msgs;
            msgs = new ArrayList<>();
        }
        return tmpMsgs;
    }

    public void onMessage(Message msg) {
        try {
            String msgText;
            if (msg instanceof TextMessage) {
                msgText = ((TextMessage) msg).getText();
            } else {
                msgText = msg.toString();
            }
            System.out.println("\n<Msg_Receiver> " + msgText);
            msgs.add(msgText);
            if (msgText.equalsIgnoreCase("quit")) {
                synchronized (this) {
                    quit = true;
                    this.notifyAll(); // Notify main thread to quit
                }
            }
        } catch (JMSException jmse) {
            jmse.printStackTrace();
        }
    }

    public void close() throws JMSException {
        qReceiver.close();
        qsession.close();
        qcon.close();
    }

    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        try {
            ic = getInitialContext();
            qconFactory = (QueueConnectionFactory) ic.lookup(QueueConst.JMS_CONNECTION_FACTORY_JNDI);
            qcon = qconFactory.createQueueConnection(QueueConst.JMS_USERNAME, QueueConst.JMS_PASSWORD);
            qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            queue = (Queue) ic.lookup(JMS_QUEUE_JNDI);
            qReceiver = qsession.createReceiver(queue);
            qReceiver.setMessageListener(this);
            qcon.start();
            synchronized (this) {
                while (!this.quit) {
                    try {
                        this.wait();
                    } catch (InterruptedException ie) {
                        ie.printStackTrace();
                    }
                }
            }
            this.close();
        } catch (JMSException | NamingException e) {
            e.printStackTrace();
        }
    }
}
